package com.course.rabbitmq.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import javax.annotation.Resource;

/**
 * @PROJECT_NAME: course
 * @DESCRIPTION:
 * @Author: 涂玄武
 * @DATE: 2021/6/16 16:26
 */
@Configuration
public class RabbitmqConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);

    @Autowired
    private Environment env;

    //自动设置RabbitMQ的链接工厂实例
    @Resource
    private CachingConnectionFactory connectionFactory;

    //自动设置消息监听器所在的容器工厂配置类实例
    @Resource
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 设置生产者的生产消息的ack信息回调(公共处理)
     */
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback() {
        return (correlationData, ack, cause) -> {
            //我们可以通过correlationData原始数据 来对消息进行后续处理，但是这是有个要求在于发送必须使用CorrelationData类
            if (ack) {
                log.info("消息发送成功：correlationData={},ack={},cause={}", correlationData, ack, cause);
            } else {
                log.error("消息发送失败,原因是:{}", cause);
            }
        };
    }

//    /**
//     * 发送者失败通知
//     * @return
//     */
//    @Bean
//    public RabbitTemplate.ReturnsCallback returnsCallback(){
//        return returnedMessage -> {
//            log.info("发送者路由失败，请检查路由：{}",returnedMessage);
//        };
//    }

    /**
     * 发送者失败通知
     * @return
     */
    @Bean
    public RabbitTemplate.ReturnCallback returnsCallback(){
        return (message, i, s, s1, s2) -> log.info("发送者路由失败，请检查路由：{},{},{},{},{}",message,i,s,s1,s2);
    }

    /**
     * Rabbitmq自定义注入模板操作组件
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        //若使用confirm-callback ，必须要配置publisherConfirms为true
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        //若使用return-callback，必须要配置publisherReturns为true
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //设置生产者的生产消息的ack信息回调(公共处理)
        rabbitTemplate.setConfirmCallback(confirmCallback());
        //当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法
        //将消息返还给生产者;
        //当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;
        //通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，否则就将消息return给发送者
        rabbitTemplate.setMandatory(true);
        //发送者失败通知
        rabbitTemplate.setReturnCallback(returnsCallback());
//        rabbitTemplate.setReturnsCallback(returnsCallback());
        return rabbitTemplate;
    }

    //单实例消费者
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        //定义消息监听器所在的容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置容器工厂所用的实例
        factory.setConnectionFactory(connectionFactory);
        //设置消息在传输中的格式，在这里采用JSON的格式进行传输
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置并发消费者实例的初始数量为1
        factory.setConcurrentConsumers(1);
        //设置并发消费者实例的最大数量为1
        factory.setMaxConcurrentConsumers(1);
        //设置并发消费者实例中每个实例拉取的消息数量为1
        factory.setPrefetchCount(1);
        return factory;
    }

    //多实例消费者
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //确认消费模式-NONE
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.concurrency", int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.max-concurrency", int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.simple.prefetch", int.class));
        return factory;
    }
}
